package com.tianmu.connector.server;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.tianmu.connector.server.config.BinlogConnectorConfig;
import com.tianmu.connector.server.factory.IClientFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Starter
 * @author tianmu
 */
public class BinlogConnectorStarter {
    private final Logger log = LoggerFactory.getLogger(BinlogConnectorStarter.class);

    private BinlogConnectorConfig binlogConnectorConfig;

    public void setBinlogConnectorConfig(BinlogConnectorConfig binlogConnectorConfig) {
        this.binlogConnectorConfig = binlogConnectorConfig;
    }

    /**
     * main start method
     */
    public void start() throws BinlogConnectorException {
        if (binlogConnectorConfig.getDistributedHandler() != null) {
            binlogConnectorConfig.getDistributedHandler().start(binlogConnectorConfig);
        } else {
            singleStart();
        }
    }

    private void singleStart() {
        //新建工厂对象
        IClientFactory binaryLogClientFactory = binlogConnectorConfig.getClientFactory();
        binaryLogClientFactory.setPositionHandler(binlogConnectorConfig.getPositionHandler());
        binaryLogClientFactory.setLifeCycleFactory(binlogConnectorConfig.getLifeCycleFactory());

        //生成全部client
        List<BinaryLogClient> binaryLogClientList = new ArrayList<>();
        binlogConnectorConfig.getSyncConfigMap().forEach((key, syncConfig) -> {
            try {
                binaryLogClientList.add(binaryLogClientFactory.getClient(syncConfig));
            } catch (BinlogConnectorException e) {
                log.error(e.getMessage(), e);
            }
        });

        //执行
        binaryLogClientList.forEach(binaryLogClient -> new Thread(() -> {
            try {
                binaryLogClient.setHeartbeatInterval(10 * 1000L);
                binaryLogClient.connect();
            } catch (IOException e) {
                log.error("binaryLogClient connect error! client={}", binaryLogClient);
            }
        }).start());
    }
}
